<?php

/**
 * Created by PhpStorm.
 * User: asus
 * Date: 2017/4/26
 * Time: 16:59
 */
set_time_limit(0);

class EpollSocketServer
{
    private static $socket;
    private static $connections;
    private static $buffers;
    private static $chats;
    private static $auth;
    private static $heartbeats;
    private static $uid;

    function EpollSocketServer($port)
    {
        global $errno, $errstr;

        if (!extension_loaded('libevent')) {
            die("Please install libevent extension firstly/n");
        }

        if ($port < 1024) {
            die("Port must be a number which bigger than 1024/n");
        }

        $socket_server = stream_socket_server("tcp://0.0.0.0:{$port}", $errno, $errstr);
        if (!$socket_server) die("$errstr ($errno)");

        stream_set_blocking($socket_server, 0); // 非阻塞

        $base = event_base_new();
        $event = event_new();
        event_set($event, $socket_server, EV_READ | EV_PERSIST, array(__CLASS__, 'ev_accept'), $base);
        event_base_set($event, $base);
        event_add($event);
        event_base_loop($base);

        self::$connections = array();
        self::$buffers = array();
        self::$chats = array();
        self::$auth = array();
        self::$heartbeats = array();
        self::$uid = array();
    }

    function ev_accept($socket, $flag, $base)
    {
        static $id = 0;

        $connection = stream_socket_accept($socket);
        stream_set_blocking($connection, 0);

        $id++; // increase on each accept

        $buffer = event_buffer_new($connection, array(__CLASS__, 'ev_read'), array(__CLASS__, 'ev_write'),
            array(__CLASS__, 'ev_error'), $id);
        event_buffer_base_set($buffer, $base);
        event_buffer_timeout_set($buffer, 30, 30);
        event_buffer_watermark_set($buffer, EV_READ, 0, 0xffffff);
        event_buffer_priority_set($buffer, 600);
        event_buffer_enable($buffer, EV_READ | EV_PERSIST);

        // we need to save both buffer and connection outside
        self::$connections[$id] = $connection;
        self::$buffers[$id] = $buffer;
        self::$heartbeats[$id] = 0;
        self::$uid[$id] = 0;
    }

    function ev_error($buffer, $error, $id)
    {
//        if(self::$heartbeats[$id] == 1){    //只有主播进入
//            $this->curlGet("http://esm.zbxinmeiti.com/Wap/Live/appEndStream?uid=".self::$uid[$id].
//                "&topic_id=".self::$chats[$id]."&socket=1");
//            $data = json_encode(array('msgType' => 7, 'liveStreamStatus' => 3));
//            $data = $this->frame($data);
//            foreach (self::$connections as $key => $conn) {
//                if (self::$chats[$id] != self::$chats[$key]) {
//                    continue;
//                } else {
//                    fwrite($conn, $data);
//                }
//            }
//        }
        event_buffer_disable(self::$buffers[$id], EV_READ | EV_WRITE);
        event_buffer_free(self::$buffers[$id]);
        fclose(self::$connections[$id]);
        unset(self::$buffers[$id], self::$connections[$id], self::$chats[$id], self::$auth[$id], self::$heartbeats[$id],
            self::$uid[$id]);
    }

    function ev_read($buffer, $id)
    {
        static $ct = 0;
        $ct_data = '';
        while ($read = event_buffer_read($buffer, 1024)) {
            $ct += strlen($read);
            $ct_data .= $read;
        }
        if (preg_match("/Sec-WebSocket-Key: (.*)\r\n/", $ct_data, $match)) {
            $key = base64_encode(sha1($match[1] . '258EAFA5-E914-47DA-95CA-C5AB0DC85B11', true));
            $upgrade = "HTTP/1.1 101 Switching Protocol\r\n" .
                "Upgrade: websocket\r\n" .
                "Connection: Upgrade\r\n" .
                "Sec-WebSocket-Accept: " . $key . "\r\n\r\n";  //必须以两个回车结尾
            event_buffer_write($buffer, $upgrade);
        } else {
            $messages = $this->decode($ct_data);
            $messageArr = json_decode($messages,true);

            $auth = isset($messageArr['auth'])?$messageArr['auth']:0;
            $topicId = isset($messageArr['topicId'])?$messageArr['topicId']:-1;
            $msgType = isset($messageArr['msgType'])?$messageArr['msgType']:0;
            $chatId = isset($messageArr['chatId'])?$messageArr['chatId']:'';
            $liveStreamStatus = isset($messageArr['liveStreamStatus'])?$messageArr['liveStreamStatus']:1;
            $uid = isset($messageArr['uid'])?$messageArr['uid']:0;

            if ($msgType == 0 || $msgType == 7 || $msgType == 9) {
                self::$chats[$id] = $topicId;
            }
            if($msgType == 0){
                self::$auth[$id] = $auth;
            }
            $data = array();
            //msgType：0人数统计 1文字 2点赞 3图文 4语音 5删除评论 6审核评论通过 7视频视频流状态 8上墙广播
            //9聊天禁言切换
            //silent：0助理发的，按钮消失 1游客发的，出现禁言按钮 2游客发的，出现恢复禁言按钮
            //status：0待审核，出现绿色√ 1已通过，出现灰色√ 3已通过，助理发的
            switch ($msgType){
                case 0:
                    $num = $socket_num = 0;
                    foreach (self::$connections as $key => $conn) {
                        if ($topicId == self::$chats[$key]) {
                            $socket_num++;
                        }
                    }
                    $res = $this->curlGet("http://esm.zbxinmeiti.com/Wap/Live/online?topic_id=".$topicId);
                    $res = json_decode($res,true);
                    if($res['status'] == 1){
                        $num = intval($res['popularity_num']) + $num;
                    }
                    $data = json_encode(array('msgType' => $msgType, 'msg' => $num, 'socket_num' => $socket_num));
                    $data = $this->frame($data);
                    break;
                case 1:     //按用户身份广播弹幕内容
                    $res = $this->curlGet("http://esm.zbxinmeiti.com/Wap/Live/getOneComment?chat_id=".$chatId);
                    $res = json_decode($res,true);
                    foreach (self::$connections as $key => $conn) {
                        if($res['comment_type'] == 3){     //助理发的消息（1观众，3助理）
                            if($topicId != self::$chats[$key]){
                                continue;
                            }else{
                                if(self::$auth[$key] == 1){     //websocket连接对象是助理
                                    $data = array('msgType' => $msgType, 'chatId' => $chatId,
                                        'nickname' => $res['nickname'], 'date' => $res['date'], 'msg' => $res['msg'],
                                        'silent' => 0, 'status' => 2, 'ask' => $res['ask']);
                                }else{
                                    $data = array('msgType' => $msgType, 'chatId' => $chatId,
                                        'nickname' => $res['nickname'], 'date' => $res['date'], 'msg' => $res['msg'],
                                        'silent' => 1, 'status' => 1, 'ask' => $res['ask']);
                                }
                                $data = $this->frame(json_encode($data));
                                fwrite($conn, $data);
                            }
                        }else{      //观众发消息
                            if($res['support_comment_check'] == 1){        //评论需要审核，只给助理发
                                if($topicId != self::$chats[$key]){
                                    continue;
                                }else{
                                    if(self::$auth[$key] == 1){     //websocket连接对象是助理
                                        $data = array('msgType' => $msgType, 'chatId' => $chatId,
                                            'nickname' => $res['nickname'], 'date' => $res['date'],
                                            'msg' => $res['msg'], 'silent' => 1, 'status' => 0, 'ask' => $res['ask']);
                                        $data = $this->frame(json_encode($data));
                                        fwrite($conn, $data);
                                    }
                                }
                            }else{          //评论不需要审核，助理和观众都发
                                if($topicId != self::$chats[$key]){
                                    continue;
                                }else{
                                    if(self::$auth[$key] == 1){     //websocket连接对象是助理
                                        $data = array('msgType' => $msgType, 'chatId' => $chatId,
                                            'nickname' => $res['nickname'], 'date' => $res['date'],
                                            'msg' => $res['msg'], 'silent' => 1, 'status' => 1, 'ask' => $res['ask']);
                                    }else{                          //websocket连接对象是观众
                                        $data = array('msgType' => $msgType, 'chatId' => $chatId,
                                            'nickname' => $res['nickname'], 'date' => $res['date'],
                                            'msg' => $res['msg'], 'silent' => 0, 'status' => 1, 'ask' => $res['ask']);
                                    }
                                    $data = $this->frame(json_encode($data));
                                    fwrite($conn, $data);
                                }
                            }
                        }
                    }
                    break;
                case 2:
                    $this->curlGet("http://esm.zbxinmeiti.com/Wap/Live/praise?topic_id=".$topicId);
                    $data = json_encode(array('msgType' => $msgType));
                    $data = $this->frame($data);
                    break;
                case 3:
                    $res = $this->curlGet("http://esm.zbxinmeiti.com/Wap/Live/getOneComment?chat_id=".$chatId);
                    $res = json_decode($res,true);
                    $data = json_encode(array('msgType' => $msgType, 'chatId' => $chatId,
                        'nickname' => $res['nickname'],  'date' => $res['date'], 'msg' => $res['msg'],
                        'images' => $res['attach']));
                    $data = $this->frame($data);
                    break;
                case 4:
                    $res = $this->curlGet("http://esm.zbxinmeiti.com/Wap/Live/getOneComment?chat_id=".$chatId);
                    $res = json_decode($res,true);
                    $data = json_encode(array('msgType' => $msgType, 'chatId' => $chatId,
                        'nickname' => $res['nickname'], 'date' => $res['date'], 'voice' => $res['attach'],
                        'record_time' => $res['record_time']));
                    $data = $this->frame($data);
                    break;
                case 5:
                    $data = json_encode(array('msgType' => $msgType, 'chatId' => $chatId));
                    $data = $this->frame($data);
                    break;
                case 6:         //按照用户身份广播，自己不广播，其他助理和观众广播，将页面上的待审核状态改变
                    $res = $this->curlGet("http://esm.zbxinmeiti.com/Wap/Live/getOneComment?chat_id=".$chatId);
                    $res = json_decode($res,true);
                    foreach (self::$connections as $key => $conn) {
                        if( $key == $id || $topicId != self::$chats[$key]){
                            continue;
                        }else{
                            if(self::$auth[$key] == 1){     //websocket连接对象是助理
                                $data = array('msgType' => $msgType, 'chatId' => $chatId, 'msg' => $res['msg'],
                                    'nickname' => $res['nickname'], 'date' => $res['date'], 'silent' => 0,
                                    'status' => 1);
                            }else{      //websocket连接对象是观众
                                $data = array('msgType' => $msgType, 'chatId' => $chatId, 'msg' => $res['msg'],
                                    'nickname' => $res['nickname'], 'date' => $res['date'], 'silent' => 1,
                                    'status' => 1);
                            }
                            $data = $this->frame(json_encode($data));
                            fwrite($conn, $data);
                        }
                    }
                    break;
                case 7:
                    self::$heartbeats[$id] = 1;
                    self::$uid[$id] = $uid;
                    $data = json_encode(array('msgType' => $msgType, 'liveStreamStatus' => $liveStreamStatus));
                    $data = $this->frame($data);
                    if($liveStreamStatus == 3){
                        $this->curlGet("http://esm.zbxinmeiti.com/Wap/Live/appEndStream?uid=".self::$uid[$id].
                            "&topic_id=".self::$chats[$id]);
                    }
                    break;
                case 8:
                    $res = $this->curlGet("http://esm.zbxinmeiti.com/Wap/Live/getOneComment?chat_id=".$chatId);
                    $res = json_decode($res,true);
                    $data = json_encode(array('msgType' => $msgType, 'chatId' => $chatId,
                        'nickname' => $res['nickname'], 'date' => $res['date'], 'msg' => $res['msg'],
                        'reply' => $res['reply']));
                    $data = $this->frame($data);
                    break;
                case 9:     //给游客发送切换消息
                    $res = $this->curlGet("http://esm.zbxinmeiti.com/Wap/Live/getSupportCommentAll?topic_id=".$topicId);
                    $res = json_decode($res,true);
                    foreach (self::$connections as $key => $conn) {
                        if($topicId != self::$chats[$key]){
                            continue;
                        }else{
                            if(self::$auth[$key] == 0) {     //websocket连接对象是观众
                                $data = array('msgType' => $msgType, 'support_comment_all' => $res["support_comment_all"]);
                            }else{
                                $data = array('msgType' => $msgType, 'support_comment_all' => 1);
                            }
                            $data = $this->frame(json_encode($data));
                            fwrite($conn, $data);
                        }
                    }
                    break;
                default:
                    break;
            }
            if($msgType != 1 && $msgType != 6 && $msgType != 9){      //已经根据身份广播过了
                foreach (self::$connections as $key => $conn) {
                    if ($topicId != self::$chats[$key]) {
                        continue;
                    } else {
                        fwrite($conn, $data);
                    }
                }
            }
        }
    }

    function ev_write($buffer, $id)
    {

        /*echo $buffer;
        if (preg_match("/Sec-WebSocket-Key: (.*)\r\n/", $buffer, $match)) {
            $key = base64_encode(sha1($match[1] . '258EAFA5-E914-47DA-95CA-C5AB0DC85B11', true));
            $upgrade = "HTTP/1.1 101 Switching Protocol\r\n" .
                "Upgrade: websocket\r\n" .
                "Connection: Upgrade\r\n" .
                "Sec-WebSocket-Accept: " . $key . "\r\n\r\n";  //必须以两个回车结尾
            event_buffer_write($buffer, $upgrade);
        }*/
    }

    public function frame($msg)
    {
        $frame = array();
        $frame[0] = '81';
        $len = strlen($msg);
        if ($len < 126) {
            $frame[1] = $len < 16 ? '0' . dechex($len) : dechex($len);
        } else if ($len < 65025) {
            $s = dechex($len);
            $frame[1] = '7e' . str_repeat('0', 4 - strlen($s)) . $s;
        } else {
            $s = dechex($len);
            $frame[1] = '7f' . str_repeat('0', 16 - strlen($s)) . $s;
        }
        $frame[2] = $this->ord_hex($msg);
        $data = implode('', $frame);
        return pack("H*", $data);
    }

    private function ord_hex($data)
    {
        $msg = '';
        $l = strlen($data);
        for ($i = 0; $i < $l; $i++) {
            $msg .= dechex(ord($data{$i}));
        }
        return $msg;
    }

    public function decode($buffer)
    {
        $len = $masks = $data = $decoded = null;
        $len = ord($buffer[1]) & 127;
        if ($len === 126) {
            $masks = substr($buffer, 4, 4);
            $data = substr($buffer, 8);
        } else if ($len === 127) {
            $masks = substr($buffer, 10, 4);
            $data = substr($buffer, 14);
        } else {
            $masks = substr($buffer, 2, 4);
            $data = substr($buffer, 6);
        }
        for ($index = 0; $index < strlen($data); $index++) {
            $decoded .= $data[$index] ^ $masks[$index % 4];
        }
        return $decoded;
    }

    public static function check_utf8($str)
    {
        $len = strlen($str);
        for ($i = 0; $i < $len; $i++) {
            $c = ord($str[$i]);
            if ($c > 128) {
                if (($c > 247)) return false;
                elseif ($c > 239) $bytes = 4;
                elseif ($c > 223) $bytes = 3;
                elseif ($c > 191) $bytes = 2;
                else return false;
                if (($i + $bytes) > $len) return false;
                while ($bytes > 1) {
                    $i++;
                    $b = ord($str[$i]);
                    if ($b < 128 || $b > 191) return false;
                    $bytes--;
                }
            }
        }
        return true;
    }

    private function curlGet($url){
        $ch = curl_init();
        curl_setopt($ch, CURLOPT_URL, $url);
        curl_setopt($ch, CURLOPT_RETURNTRANSFER, 1);
        curl_setopt($ch, CURLOPT_TIMEOUT, 500);
        curl_setopt($ch, CURLOPT_SSL_VERIFYPEER, true);
        curl_setopt($ch, CURLOPT_SSL_VERIFYHOST, 2);
        $out = curl_exec($ch);
        curl_close($ch);
        return $out;
    }

    private function curlPost($url, $post_string){
        $ch = curl_init();
        curl_setopt($ch, CURLOPT_URL, $url);
        curl_setopt($ch, CURLOPT_POSTFIELDS, $post_string);
        curl_setopt($ch, CURLOPT_RETURNTRANSFER, true);
        curl_setopt($ch, CURLOPT_USERAGENT, "CURL post");
        $output = curl_exec($ch);
        curl_close($ch);
        return $output;
    }
}

new EpollSocketServer(7000);